package com.example.flinkcourse.lesson3.transform;

import com.example.flinkcourse.lesson3.model.Transaction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 交易处理器
 * 使用 DataStream API 处理交易数据
 */
public class TransactionProcessor {
    private static final Logger LOG = LoggerFactory.getLogger(TransactionProcessor.class);

    /**
     * 处理交易流
     * @param inputStream 输入交易流
     * @return 处理后的交易流
     */
    public static DataStream<Transaction> processTransactions(DataStream<Transaction> inputStream) {
        LOG.info("Processing transactions with DataStream API...");
        
        return inputStream
            // 过滤无效交易
            .filter(transaction -> {
                boolean isValid = transaction != null && transaction.getAmount() > 0;
                if (!isValid) {
                    LOG.warn("Invalid transaction received: {}", transaction);
                }
                return isValid;
            })
            
            // 添加处理时间
            .map(transaction -> {
                transaction.setTimestamp(System.currentTimeMillis());
                LOG.debug("Added processing time to transaction: {}", transaction.getId());
                return transaction;
            })
            
            // 按账户分组
            .keyBy(Transaction::getAccountId)
            
            // 窗口处理
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            
            // 聚合处理
            .aggregate(new TransactionAggregator())
            
            // 设置并行度
            .setParallelism(4);
    }
}

/**
 * 交易聚合器
 * 用于在窗口内聚合交易数据
 */
class TransactionAggregator implements AggregateFunction<Transaction, Transaction, Transaction> {
    @Override
    public Transaction createAccumulator() {
        return new Transaction();
    }

    @Override
    public Transaction add(Transaction transaction, Transaction accumulator) {
        // 更新聚合结果
        if (accumulator.getId() == null) {
            accumulator.setId(transaction.getId());
            accumulator.setAccountId(transaction.getAccountId());
            accumulator.setAmount(transaction.getAmount());
            accumulator.setTimestamp(transaction.getTimestamp());
            accumulator.setType(transaction.getType());
            accumulator.setProperties(transaction.getProperties());
        } else {
            // 累加金额
            accumulator.setAmount(accumulator.getAmount() + transaction.getAmount());
            // 更新最新时间戳
            if (transaction.getTimestamp() > accumulator.getTimestamp()) {
                accumulator.setTimestamp(transaction.getTimestamp());
            }
        }
        return accumulator;
    }

    @Override
    public Transaction getResult(Transaction accumulator) {
        return accumulator;
    }

    @Override
    public Transaction merge(Transaction a, Transaction b) {
        // 合并两个聚合结果
        if (a.getTimestamp() > b.getTimestamp()) {
            a.setAmount(a.getAmount() + b.getAmount());
            return a;
        } else {
            b.setAmount(a.getAmount() + b.getAmount());
            return b;
        }
    }
} 